package com.zc.manager.device.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ruoyi.common.constant.HttpStatus;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.PageUtils;
import com.ruoyi.common.utils.bean.BeanValidators;
import com.zc.common.core.foreignkeyservice.AssociatedService;
import com.zc.common.core.foreignkeyservice.ForeignKeyService;
import com.zc.common.core.foreignkeyservice.LinkModel;
import com.zc.component.common.utils.ErrorUtils;
import com.zc.component.gateway.annotation.Subscribe;
import com.zc.core.Value;
import com.zc.core.Values;
import com.zc.core.device.*;
import com.zc.core.enums.ErrorCode;
import com.zc.core.exception.DeviceOperationException;
import com.zc.core.message.*;
import com.zc.core.message.function.FunctionInvokeMessageReply;
import com.zc.core.message.property.ReadPropertyMessageReply;
import com.zc.core.message.property.WritePropertyMessageReply;
import com.zc.core.metadata.ConfigMetadata;
import com.zc.core.metadata.DeviceMetadata;
import com.zc.core.server.DeviceGatewayContext;
import com.zc.core.utils.CyclicDependencyChecker;
import com.zc.core.utils.Reactors;
import com.zc.manager.device.constant.IotRedisKey;
import com.zc.manager.device.domain.IotDevice;
import com.zc.manager.device.domain.IotDeviceTags;
import com.zc.manager.device.domain.IotProduct;
import com.zc.manager.device.entity.DeviceStateInfo;
import com.zc.manager.device.enums.DeviceState;
import com.zc.manager.device.mapper.IotDeviceMapper;
import com.zc.manager.device.mapper.IotDeviceTagsMapper;
import com.zc.manager.device.response.DeviceDeployResult;
import com.zc.manager.device.response.DeviceDetail;
import com.zc.manager.device.service.DeviceConfigMetadataManager;
import com.zc.manager.device.service.IIotDeviceService;
import com.zc.manager.device.service.IIotDeviceTagsService;
import com.zc.manager.device.service.IIotProductService;
import com.zc.web.bean.FastBeanCopier;
import com.zc.web.crud.EntityDeletedEvent;
import com.zc.web.exception.BusinessException;
import com.zc.web.id.IDGenerator;
import com.zc.web.exception.NotFoundException;
import lombok.SneakyThrows;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.reactivestreams.Publisher;
import org.springframework.context.event.EventListener;
import org.springframework.data.util.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;
import reactor.util.function.Tuples;

import javax.annotation.Nonnull;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.validation.Validator;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 产品Service业务层处理
 *
 * @author zhaoxianglong
 * @date 2022-11-10
 */
@Service
public class IotDeviceServiceImpl
        extends ServiceImpl<IotDeviceMapper, IotDevice>
        implements IIotDeviceService, ForeignKeyService {

    @Resource(name = "iotProductServiceImpl")
    private AssociatedService associatedService;

    @Resource
    private IIotProductService productService;

    @Resource
    private IIotDeviceTagsService deviceTagsService;

    @Resource
    private DeviceRegistry registry;

    @Resource
    private IotDeviceTagsMapper iotDeviceTagsMapper;

    @Resource
    private DeviceConfigMetadataManager metadataManager;

    @Resource
    private RedisCache redisCache;

    @Resource
    protected Validator validator;

    @PostConstruct
    public void init() {
        // 注册外键
        registerForeignKey();
    }


    /**
     * 服务连接事件
     *
     * @param deviceGatewayContext 设备网关上下文
     * @return
     */
    @Subscribe("/server/*/connected")
    public Mono<Void> serverConnected(DeviceGatewayContext deviceGatewayContext) {
        return Mono.fromRunnable(() -> {

            InetSocketAddress address = deviceGatewayContext.getRemoteAddress();

            String deviceId = redisCache.getCacheMapValue(IotRedisKey.DEVICE_ADDRESS, String.valueOf(address));

            if (deviceId == null) {

                deviceId = redisCache.getCacheMapValue(IotRedisKey.DEVICE_ADDRESS, "/" + address.getHostString());

                if (deviceId == null) {
                    return;
                }
            }

            DeviceOnlineMessage onlineMessage = new DeviceOnlineMessage();
            onlineMessage.setDeviceId(deviceId);
            onlineMessage.setTimestamp(System.currentTimeMillis());
            deviceGatewayContext.onMessage(onlineMessage).subscribe();
        });
    }


    //*********************************设备增删改查******************************************

    /**
     * 添加设备信息
     *
     * @param iotDevice 设备信息
     * @return 操作结果
     */
    @Override
    public boolean addDevice(IotDevice iotDevice) {

        String id = iotDevice.getId();

        // 检查设备id是否重复
        IotDevice device = getById(id);

        if (device != null) {
            throw new ServiceException("设备ID[" + id + "]已存在", HttpStatus.BAD_REQUEST);
        }

        String productId = iotDevice.getProductId();

        // 检查产品是否存在
        if (productId != null) {
            IotProduct iotProduct = productService.selectIotProductById(productId);
            // 产品不存在
            if (iotProduct == null) {
                throw new ServiceException("所属产品[" + productId + "]不存在", HttpStatus.BAD_REQUEST);
            }
        }

        String address = iotDevice.getAddress();

        // 检查设备地址是否重复
        if (StringUtils.hasText(address)) {
            LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(IotDevice::getAddress, address);
            List<IotDevice> iotDeviceList = list(queryWrapper);
            if (iotDeviceList != null && !iotDeviceList.isEmpty()) {
                throw new ServiceException("设备地址[" + address + "]重复", HttpStatus.BAD_REQUEST);
            }
        }

        String parentId = iotDevice.getParentId();

        // 校验父设备
        if (StringUtils.hasLength(parentId)) {
            verifyParentDevice(productId, parentId);
        }


        iotDevice.setCreateTime(new Date());
        iotDevice.setDeviceState(IotDevice.NOT_ACTIVE);
        return save(iotDevice);
    }

    /**
     * 修改设备信息
     *
     * @param iotDevice 设备信息
     * @return 操作结果
     */
    @Override
    public boolean editDevice(IotDevice iotDevice) {

        String id = iotDevice.getId();

        // 检查设备id是否重复
        IotDevice device = getById(id);

        if (device == null) {
            throw new ServiceException("设备ID[" + id + "]不存在", HttpStatus.BAD_REQUEST);
        }

        if (!IotDevice.NOT_ACTIVE.equals(device.getDeviceState())) {
            throw new ServiceException("不允许修改已启用的设备", HttpStatus.FORBIDDEN);
        }

        String productId = iotDevice.getProductId();

        // 检查产品是否存在
        if (productId != null) {
            IotProduct iotProduct = productService.selectIotProductById(productId);
            // 产品不存在
            if (iotProduct == null) {
                throw new ServiceException("所属产品[" + productId + "]不存在", HttpStatus.BAD_REQUEST);
            }
        }

        String address = iotDevice.getAddress();

        // 检查设备地址是否重复
        if (StringUtils.hasText(address)) {
            LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(IotDevice::getAddress, address);
            List<IotDevice> iotDeviceList = list(queryWrapper);
            if (iotDeviceList != null && !iotDeviceList.isEmpty() && !id.equals(iotDeviceList.get(0).getId())) {
                throw new ServiceException("设备地址[" + address + "]重复", HttpStatus.BAD_REQUEST);
            }
        }

        String parentId = iotDevice.getParentId();

        // 校验父设备
        if (StringUtils.hasLength(parentId)) {
            verifyParentDevice(productId, parentId);
        }

        iotDevice.setUpdateTime(new Date());
        return updateById(iotDevice);
    }

    private void verifyParentDevice(String productId, String parentId) {

        if (productId == null || parentId == null) {
            return;
        }

        // 检查当前设备类型
        IotProduct iotProduct = productService.selectIotProductById(productId);
        // 当前设备是否是个网关设备
        if (iotProduct != null && IotProduct.GATEWAY.equals(iotProduct.getDeviceType())) {
            throw new ServiceException("当前设备为网关设备，不允许添加父设备" , HttpStatus.BAD_REQUEST);
        }

        IotDevice parentDevice = getById(parentId);
        // 父设备不存在
        if (parentDevice == null) {
            throw new ServiceException("网关设备（父设备）[" + parentId + "]不存在" , HttpStatus.NOT_FOUND);
        }

        IotProduct parentProduct = productService.selectIotProductById(parentDevice.getProductId());

        if(parentProduct != null && !IotProduct.GATEWAY.equals(parentProduct.getDeviceType())) {
            throw new ServiceException("父设备不是一个网关设备" , HttpStatus.NOT_FOUND);
        }
    }

    /**
     * 删除设备
     *
     * @param ids 设备ID
     * @return 操作结果
     */
    @Override
    @Transactional
    public boolean removeDevice(List<String> ids) {

        // 根据 ids 批量查询设备信息
        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.in(IotDevice::getId, ids);
        List<IotDevice> list = list(queryWrapper);

        // 离线或者在线设备不允许删除
        for (IotDevice iotDevice : list) {
            if (IotDevice.OFFLINE.equals(iotDevice.getDeviceState()) || IotDevice.ONLINE.equals(iotDevice.getDeviceState())) {
                return false;
            }
        }

        // 删除设备标签
        for (IotDevice iotDevice : list) {
            LambdaQueryWrapper<IotDeviceTags> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(IotDeviceTags::getDeviceId, iotDevice.getId());
            deviceTagsService.remove(lambdaQueryWrapper);
        }

        return removeByIds(ids);
    }

    /**
     * 分页查询设备列表
     *
     * @param iotDevice 参数
     * @return 结果
     */
    @Override
    public List<IotDevice> pageDevice(IotDevice iotDevice) {
        // 分页
        PageUtils.startPage();
        return list(deviceQueryWrapper(iotDevice));
    }

    /**
     * 获取设备列表
     *
     * @param iotDevice 参数
     * @return 结果
     */
    @Override
    public List<IotDevice> listDevice(IotDevice iotDevice) {
        return list(deviceQueryWrapper(iotDevice));
    }

    public LambdaQueryWrapper<IotDevice> deviceQueryWrapper(IotDevice iotDevice) {

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();

        // 设备id
        if (StringUtils.hasText(iotDevice.getId())) {
            queryWrapper.eq(IotDevice::getId, iotDevice.getId());
        }

        // 设备名称
        if (StringUtils.hasText(iotDevice.getDeviceName())) {
            queryWrapper.like(IotDevice::getDeviceName, iotDevice.getDeviceName());
        }

        // 设备编号
        if (StringUtils.hasText(iotDevice.getDeviceCode())) {
            queryWrapper.like(IotDevice::getDeviceCode, iotDevice.getDeviceCode());
        }

        // 产品id
        if (StringUtils.hasText(iotDevice.getProductId())) {
            queryWrapper.eq(IotDevice::getProductId, iotDevice.getProductId());
        }

        // 设备状态
        if (StringUtils.hasText(iotDevice.getDeviceState())) {
            queryWrapper.eq(IotDevice::getDeviceState, iotDevice.getDeviceState());
        }

        // 父设备
        if (StringUtils.hasText(iotDevice.getParentId())) {
            queryWrapper.eq(IotDevice::getParentId, iotDevice.getParentId());
        }

        // 设备类型
        if (StringUtils.hasText(iotDevice.getDeviceType())) {

            IotProduct iotProduct = new IotProduct();
            iotProduct.setDeviceType(iotDevice.getDeviceType());

            List<IotProduct> iotProducts = productService.selectIotProductList(iotProduct);

            if (iotProducts != null && !iotProducts.isEmpty()) {
                List<String> productIds = iotProducts.stream().map(IotProduct::getId).collect(Collectors.toList());
                queryWrapper.in(IotDevice::getProductId, productIds);
            }
        }

        // 设备地址
        if (StringUtils.hasText(iotDevice.getAddress())) {
            queryWrapper.eq(IotDevice::getAddress, iotDevice.getAddress());
        }

        return queryWrapper;
    }

    /**
     * 导入设备数据
     *
     * @param deviceList    设备数据列表
     * @param updateSupport 是否更新支持，如果已存在，则进行更新数据
     * @return 结果
     */
    @Override
    public String importDevice(List<IotDevice> deviceList, boolean updateSupport) {
        if (deviceList == null || deviceList.isEmpty())
        {
            throw new ServiceException("导入设备数据不能为空！");
        }
        int successNum = 0;
        int failureNum = 0;
        StringBuilder successMsg = new StringBuilder();
        StringBuilder failureMsg = new StringBuilder();
        for (IotDevice device : deviceList)
        {
            try
            {
                if (updateSupport)
                {
                    BeanValidators.validateWithException(validator, device);
                    this.editDevice(device);
                    successNum++;
                    successMsg.append("<br/>").append(successNum).append("、设备 ").append(device.getId()).append(" 更新成功");
                }
                else
                {
                    BeanValidators.validateWithException(validator, device);
                    this.addDevice(device);
                    successNum++;
                    successMsg.append("<br/>").append(successNum).append("、设备 ").append(device.getId()).append(" 导入成功");
                }
            }
            catch (Exception e)
            {
                failureNum++;
                String msg = "<br/>" + failureNum + "、设备 " + device.getId() + " 导入失败：";
                failureMsg.append(msg).append(e.getMessage());
                log.error(msg, e);
            }
        }
        if (failureNum > 0)
        {
            failureMsg.insert(0, "很抱歉，导入失败！共 " + failureNum + " 条数据格式不正确，错误如下：");
            throw new ServiceException(failureMsg.toString());
        }
        else
        {
            successMsg.insert(0, "恭喜您，数据已全部导入成功！共 " + successNum + " 条，数据如下：");
        }
        return successMsg.toString();
    }

    /**
     * 根据id查询设备信息
     *
     * @param id 设备ID
     * @return 设备信息
     */
    @Override
    public IotDevice getDevice(String id) {
        return getById(id);
    }

    /**
     * 根据产品id查询设备信息
     *
     * @param productId 产品id
     * @return 查询结果
     */
    @Override
    public Flux<IotDevice> getDeviceCache(String productId) {
        return registry
                .getProduct(productId)
                .flatMapMany(DeviceProductOperator::getDevices)
                .flatMap(deviceOperator ->
                        Mono.defer(() ->
                                Mono.zip(deviceOperator.getMetadata(), deviceOperator.getSelfConfigs("deviceName"))
                                    .map(tuple2 -> {
                                        Values values = tuple2.getT2();
                                        String deviceName = values.getValue("deviceName").map(Value::asString).orElse(null);
                                        IotDevice iotDevice = new IotDevice();
                                        iotDevice.setId(deviceOperator.getId());
                                        iotDevice.setDeviceName(deviceName);
                                        iotDevice.setProductId(productId);
                                        iotDevice.setDeriveMetadata(tuple2.getT1().toJson().toJSONString());
                                        return iotDevice;
                                    })
                        )
                );
    }

    /**
     * 获取设备详情信息
     *
     * @param deviceId 设备id
     * @return 设备详情信息
     */
    public Mono<DeviceDetail> getDeviceDetail(String deviceId) {
        return Mono.just(getById(deviceId))
                .zipWhen(device -> Mono.just(productService.selectIotProductById(device.getProductId())))//合并型号
                .zipWith(Mono.just(iotDeviceTagsMapper.selectIotDeviceTagsByDeviceId(deviceId)) //合并标签
                        , (left, right) -> Tuples.of(left.getT2(), left.getT1(), right))
                .flatMap(tp3 -> createDeviceDetail(tp3.getT1(), tp3.getT2(), tp3.getT3()));
    }

    /**
     * 根据条件获取设备数量
     *
     * @param iotDevice 查询条件
     * @return 数量
     */
    @Override
    public Integer getDeviceCount(IotDevice iotDevice) {
        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();

        if (StringUtils.hasText(iotDevice.getProductId())) {
            queryWrapper.eq(IotDevice::getProductId, iotDevice.getProductId());
        }

        if (StringUtils.hasText(iotDevice.getDeviceState())) {
            queryWrapper.eq(IotDevice::getDeviceState, iotDevice.getDeviceState());
        }

        return count(queryWrapper);
    }


    //*********************************设备状态******************************************

    /**
     * 启用设备
     *
     * @param ids 设备id 集合
     * @return 操作结果
     */
    @Override
    public boolean deployDevice(List<String> ids) {

        if (ids == null || ids.isEmpty()) {
            return false;
        }

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.in(IotDevice::getId, ids);

        List<IotDevice> iotDevices = list(queryWrapper);

        if (iotDevices == null || iotDevices.isEmpty()) {
            return false;
        }

        boolean result = Objects.requireNonNull(Flux.fromIterable(iotDevices)
                .as(this::deploy)
                .singleOrEmpty()
                .block())
                .isSuccess();

        if (result) {
            iotDevices.forEach(iotDevice -> {
                String address = iotDevice.getAddress();
                if (!StringUtils.hasText(address)) {
                    return;
                }

                redisCache.setCacheMapValue(IotRedisKey.DEVICE_ADDRESS, address, iotDevice.getId());
            });
        }

        return result;

    }

    /**
     * 发布设备到设备注册中心
     *
     * @param id 设备ID
     * @return 发布结果
     */
    @Override
    public Mono<DeviceDeployResult> deploy(String id) {
        return Mono.just(getById(id))
                .flux()
                .as(this::deploy)
                .singleOrEmpty();
    }

    private void addChildren(@Nonnull IotDevice iotDevice) {

        if (StringUtils.hasText(iotDevice.getParentId())) {
            return;
        }

        IotProduct iotProduct = productService.selectIotProductById(iotDevice.getProductId());

        if (!IotProduct.GATEWAY.equals(iotProduct.getDeviceType())) {
            return;
        }

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(IotDevice::getParentId, iotDevice.getId());

        List<IotDevice> iotDeviceList = list(queryWrapper);

        if (iotDeviceList == null) {
            return;
        }

        Set<String> childrenId = iotDeviceList
                .stream()
                .filter(device -> !IotDevice.NOT_ACTIVE.equals(device.getDeviceState()))
                .map(IotDevice::getId)
                .collect(Collectors.toSet());

        iotDevice.setChildrenId(childrenId);
    }


    /**
     * 激活全部设备
     *
     * @return 发布结果
     */
    @Override
    public boolean deployAllDevice() {

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(IotDevice::getDeviceState, IotDevice.NOT_ACTIVE);

        List<IotDevice> iotDeviceList = list(queryWrapper);

        List<String> ids = iotDeviceList.stream().map(IotDevice::getId).collect(Collectors.toList());

        return deployDevice(ids);
    }

    /**
     * 批量发布设备到设备注册中心
     *
     * @param flux 设备实例流
     * @return 发布数量
     */
    @Override
    @Transactional
    public Flux<DeviceDeployResult> deploy(Flux<IotDevice> flux) {
        return flux
                .flatMap(instance -> {
                    addChildren(instance);
                    return registry
                        .register(instance.toDeviceInfo())
                        .flatMap(deviceOperator -> deviceOperator
                                .getState()
                                .flatMap(r -> {
                                    if (r.equals(com.zc.core.device.DeviceState.unknown) ||
                                            r.equals(com.zc.core.device.DeviceState.noActive)) {
                                        instance.setDeviceState(DeviceState.offline.getValue());
                                        return deviceOperator.putState(com.zc.core.device.DeviceState.offline);
                                    }
                                    instance.setDeviceState(DeviceState.of(r).getValue());
                                    return Mono.just(true);
                                })
                                .flatMap(success -> success ? Mono.just(deviceOperator) : Mono.empty())
                        )
                        .thenReturn(instance);})
                .buffer(50)
                .publishOn(Schedulers.single())
                .flatMap(all -> Flux
                        .fromIterable(all)
                        .groupBy(IotDevice::getDeviceState)
                        .flatMap(group -> group
                                .map(IotDevice::getId)
                                .collectList()
                                .flatMap(list -> {
                                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                                    updateWrapper
                                            .set(IotDevice::getDeviceState, group.key())
                                            .set(IotDevice::getRegistryTime, new Date())
                                            .in(IotDevice::getId, list);
                                    return Mono.just(update(updateWrapper))
                                            .map(r -> DeviceDeployResult.success(list.size()))
                                            .onErrorResume(err -> Mono.just(DeviceDeployResult.error(err.getMessage())));
                                })))

                ;


    }


    /**
     * 停用设备
     *
     * @param ids 设备id 集合
     * @return 操作结果
     */
    @Override
    public boolean undeployDevice(List<String> ids) {

        if (ids == null || ids.isEmpty()) {
            return false;
        }

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.in(IotDevice::getId, ids);

        List<IotDevice> iotDevices = list(queryWrapper);

        if (iotDevices == null || iotDevices.isEmpty()) {
            return false;
        }

        boolean result = Objects.requireNonNull(Flux.fromIterable(iotDevices)
                        .map(IotDevice::getId)
                        .as(this::unregisterDevice)
                        .block());


        if (result) {
            iotDevices.forEach(iotDevice -> {
                String address = iotDevice.getAddress();
                if (!StringUtils.hasText(address)) {
                    return;
                }

                redisCache.delCacheMapValue(IotRedisKey.DEVICE_ADDRESS, address);
            });
        }

        return result;
    }

    /**
     * 激活全部设备
     *
     * @return 发布结果
     */
    @Override
    public boolean undeployAllDevice() {

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.ne(IotDevice::getDeviceState, IotDevice.NOT_ACTIVE);

        List<IotDevice> iotDeviceList = list(queryWrapper);

        List<String> ids = iotDeviceList.stream().map(IotDevice::getId).collect(Collectors.toList());

        return undeployDevice(ids);
    }

    /**
     * 断开设备连接
     *
     * @param id 设备id
     * @return 操作结果
     */
    @Override
    public boolean disconnectDevice(String id) {
        Boolean result = registry
                .getDevice(id)
                .flatMapMany(DeviceOperator::disconnect)
                .singleOrEmpty().block();
        return Objects.requireNonNull(result);
    }

    /**
     * 同步设备状态
     *
     * @param ids 设备id集合
     * @return 完成同步个数
     */
    @Override
    @Transactional
    public int syncDeviceState(List<String> ids) {

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.in(IotDevice::getId, ids);

        Flux<Integer> count = Flux.fromIterable(list(queryWrapper))
                .map(IotDevice::getId)
                .buffer(200)
                .publishOn(Schedulers.single())
                .concatMap(flux -> syncStateBatch(Flux.just(flux), true).map(List::size))
                .defaultIfEmpty(0);

        Integer result = count.blockFirst();

        if (result == null) {
            result = 0;
        }

        return result;

    }

    /**
     * 同步全部设备状态
     *
     * @return
     */
    @Override
    public int syncAllDeviceState() {
        Flux<Integer> count = Flux.fromIterable(list())
                .map(IotDevice::getId)
                .buffer(200)
                .publishOn(Schedulers.single())
                .concatMap(flux -> syncStateBatch(Flux.just(flux), true).map(List::size))
                .defaultIfEmpty(0);
        Integer result = count.blockFirst();

        if (result == null) {
            result = 0;
        }

        return result;
    }

    /**
     * 批量同步设备状态
     *
     * @param batch
     * @param force
     * @return
     */
    @Override
    @Transactional
    public Flux<List<DeviceStateInfo>> syncStateBatch(Flux<List<String>> batch, boolean force) {
        return batch
                .concatMap(list -> Flux
                        .fromIterable(list)
                        .publishOn(Schedulers.parallel())
                        .flatMap(id -> registry
                                .getDevice(id)
                                .flatMap(operator -> {
                                    // 获取状态
                                    Mono<Byte> state = force
                                            ? operator
                                            .checkState()
                                            .onErrorResume(err -> operator.getState())
                                            : operator.getState();
                                    return Mono
                                            .zip(
                                                    state.defaultIfEmpty(com.zc.core.device.DeviceState.offline),//状态
                                                    Mono.just(operator.getDeviceId()), //设备id
                                                    operator.getConfig(DeviceConfigKey.isGatewayDevice).defaultIfEmpty(false)//是否为网关设备
                                            );
                                })
                                //注册中心里不存在设备就认为是未激活.
                                .defaultIfEmpty(Tuples.of(com.zc.core.device.DeviceState.noActive, id, false)))
                        .collect(Collectors.groupingBy(Tuple2::getT1))
                        .flatMapIterable(Map::entrySet)
                        .flatMap(group -> {
                            List<String> deviceIdList = group
                                    .getValue()
                                    .stream()
                                    .map(Tuple3::getT2)
                                    .collect(Collectors.toList());
                            DeviceState state = DeviceState.of(group.getKey());

                            LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                            updateWrapper
                                    .set(IotDevice::getDeviceState, state.getValue())
                                    .in(IotDevice::getId, deviceIdList);
                            return
                                    //批量修改设备状态
                                    Mono.just(update(updateWrapper))
                                            .thenReturn(group.getValue().size())
                                            .then(Mono.just(
                                                    deviceIdList
                                                            .stream()
                                                            .map(id -> DeviceStateInfo.of(id, state))
                                                            .collect(Collectors.toList())
                                            ));
                        }));
    }

    /**
     * 注销设备,取消后,设备无法再连接到服务. 注册中心也无法再获取到该设备信息.
     *
     * @param deviceId 设备ID
     * @return 注销结果
     */
    @Override
    public Mono<Boolean> unregisterDevice(String deviceId) {

        IotDevice iotDevice = getById(deviceId);
        iotDevice.setId(deviceId);
        iotDevice.setDeviceState(IotDevice.NOT_ACTIVE);
        iotDevice.setRegistryTime(null);

        String parentId = iotDevice.getParentId();

        if (!StringUtils.isEmpty(parentId)) {
            return registry
                    .getDevice(parentId)
                    .flatMap(deviceOperator -> deviceOperator.removeDeviceChildrenIdCache(deviceId))
                    .flatMap(removed -> {
                        if (removed) {
                            return registry
                                    .unregisterDevice(deviceId)
                                    .then(Mono.just(updateById(iotDevice)));
                        }
                        return Mono.just(Boolean.FALSE);
                    });
        }

        return registry
                .unregisterDevice(deviceId)
                .then(Mono.just(updateById(iotDevice)));

    }

    /**
     * 批量注销设备
     *
     * @param ids 设备ID
     * @return 注销结果
     */
    @Override
    @Transactional
    public Mono<Boolean> unregisterDevice(Publisher<String> ids) {

        return Flux.from(ids)
                .flatMap(id -> registry.unregisterDevice(id).thenReturn(id))
                .collectList()
                .flatMap(list -> {
                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper
                            .set(IotDevice::getDeviceState, IotDevice.NOT_ACTIVE)
                            .in(IotDevice::getId, list);
                    return Mono.just(update(updateWrapper));
                });
    }

    /**
     * 获取设备状态
     *
     * @param deviceId 设id
     * @return 设备状态结果
     */
    @Override
    public Mono<DeviceState> getDeviceState(String deviceId) {
        return registry
                .getDevice(deviceId)
                .flatMap(DeviceOperator::checkState)
                .flatMap(state -> {
                    DeviceState deviceState = DeviceState.of(state);

                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.set(IotDevice::getDeviceState, DeviceState.of(state));
                    updateWrapper.eq(IotDevice::getId, deviceId);
                    return Mono.just(update(updateWrapper))
                            .thenReturn(deviceState);
                })
                .defaultIfEmpty(DeviceState.notActive);
    }


    //*********************************子设备******************************************

    /**
     * 查询未绑定的子设备
     *
     * @param iotDevice 查询参数
     * @return 查询结果
     */
    @Override
    public List<IotDevice> listChildrenDevice(IotDevice iotDevice) {

        LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();

        if (iotDevice.getId() != null) {
            queryWrapper.eq(IotDevice::getId, iotDevice.getId());
        }

        if (iotDevice.getDeviceName() != null) {
            queryWrapper.like(IotDevice::getDeviceName, iotDevice.getDeviceName());
        }

        if (iotDevice.getDeviceCode() != null) {
            queryWrapper.like(IotDevice::getDeviceCode, iotDevice.getDeviceCode());
        }

        if (iotDevice.getParentId() == null) {
            queryWrapper.isNull(IotDevice::getParentId);
        }

        if (iotDevice.getProductId() != null) {
            queryWrapper.eq(IotDevice::getProductId, iotDevice.getProductId());
        }

        if (iotDevice.getDeviceType() != null) {
            IotProduct iotProduct = new IotProduct();
            iotProduct.setDeviceType(iotDevice.getDeviceType());
            List<IotProduct> iotProducts = productService.selectIotProductList(iotProduct);
            if (iotProducts == null || iotProducts.isEmpty()) {
                return new ArrayList<>();
            }
            List<String> iotProductIds = iotProducts.stream().map(IotProduct::getId).collect(Collectors.toList());
            queryWrapper.in(IotDevice::getProductId, iotProductIds);
        }

        // 分页
        PageUtils.startPage();

        return list(queryWrapper);
    }

    /**
     * 绑定子设备
     *
     * @param ids      设备ID
     * @param parentId 父设备id
     * @return 绑定结果
     */
    @Override
    @Transactional
    public Mono<Boolean> bindChildrenDevice(List<String> ids, String parentId) {

        return Mono.just(ids)
                .flatMapIterable(Function.identity())
                .flatMap(childId -> this
                        .checkCyclicDependency(childId, parentId)
                        .thenReturn(childId))
                .collectList()
                .filter(CollectionUtils::isNotEmpty)
                .flatMap(deviceIdList ->
                        Mono.defer(() -> {
                            // 更新数据库数据
                            LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                            updateWrapper
                                    .set(IotDevice::getParentId, parentId)
                                    .in(IotDevice::getId, deviceIdList);
                            return Mono.just(update(updateWrapper));
                        }).then(Flux // 更新缓存
                                .fromIterable(deviceIdList)
                                .flatMap(id -> registry
                                        .getDevice(id)
                                        .flatMap(operator -> operator.setConfig(DeviceConfigKey.parentGatewayId, parentId)))
                                .then()
                        ).then(registry
                                .getDevice(parentId)
                                .flatMap(gwOperator -> gwOperator
                                        .getProtocol()
                                        .flatMap(protocolSupport -> protocolSupport
                                                .onChildBind(gwOperator,
                                                        Flux.fromIterable(deviceIdList).flatMap(id -> registry.getDevice(id)))
                                        )
                                )
                        )
                ).then(Reactors.ALWAYS_TRUE);

    }

    /**
     * 解除绑定
     *
     * @param ids 要解除的ID集合
     * @return 处理结果
     */
    @Override
    @Transactional
    public Mono<Boolean> unbindChildrenDevice(List<String> ids) {

        return Mono.just(ids)
                .flatMapIterable(Function.identity())
                .collectList()
                .filter(CollectionUtils::isNotEmpty)
                .flatMap(deviceIdList ->
                        Mono.defer(() -> {
                            // 更新数据库数据
                            LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                            updateWrapper
                                    .set(IotDevice::getParentId, null)
                                    .in(IotDevice::getId, deviceIdList);
                            return Mono.just(update(updateWrapper));
                        }).then(Flux.fromIterable(deviceIdList)
                                        .flatMap(id -> registry
                                                .getDevice(id)
                                                .flatMap(operator -> operator.getAndRemoveConfig(DeviceConfigKey.parentGatewayId.getKey()))
                                                .map(Value::asString)
                                                .map(parentId -> registry.getDevice(parentId)
                                                        .flatMap(gwOperator -> gwOperator.getProtocol()
                                                                .flatMap(protocolSupport -> protocolSupport.onChildUnbind(gwOperator,
                                                                        Flux.from(registry.getDevice(id)))
                                                                )
                                                        )
                                                )
                                        ).then()
                        ).then(Reactors.ALWAYS_TRUE));

    }


    //*********************************设备其他配置*************************************

    /**
     * @param address 设备地址
     * @return 获取配置信息
     */
    public Mono<Map<String, Object>> getByAddressConfiguration(String address) {

        String deviceId = redisCache.getCacheMapValue(IotRedisKey.DEVICE_ADDRESS, address);

        if (deviceId == null) {
            return Mono.empty();
        }

        return Mono
                .just(deviceId)
                .flatMap(this::getByIdConfiguration);

    }

    /**
     * @param id 设备id
     * @return 获取配置信息
     */
    public Mono<Map<String, Object>> getByIdConfiguration(String id) {
        return getDeviceDetail(id)
                .map(DeviceDetail::getConfiguration);
    }


    /**
     * 更新设备其他配置
     *
     * @param id            设备id
     * @param configuration 其他配置
     * @return 更新结果
     */
    @Override
    public boolean updateConfiguration(String id, String configuration) {
        LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper
                .eq(IotDevice::getId, id)
                .set(IotDevice::getConfiguration, configuration);
        return update(updateWrapper);
    }

    /**
     * 重置设备配置
     *
     * @param id 设备id
     * @return 结果
     */
    @Override
    public Mono<Map<String, Object>> resetConfiguration(String id) {
        return Mono.just(getById(id))
                .zipWhen(device -> Mono.just(productService.selectIotProductById(device.getProductId())))
                .flatMap(tp2 -> {
                    IotProduct product = tp2.getT2();
                    IotDevice device = tp2.getT1();

                    Map<String, Object> productConfiguration = JSONObject.parseObject(product.getConfiguration());
                    Map<String, Object> deviceConfiguration = JSONObject.parseObject(device.getConfiguration());

                    return Mono
                            .defer(() -> {
                                if (MapUtils.isNotEmpty(productConfiguration)) {
                                    if (MapUtils.isNotEmpty(deviceConfiguration)) {
                                        productConfiguration
                                                .keySet()
                                                .forEach(deviceConfiguration::remove);
                                    }

                                    //重置注册中心里的配置
                                    return registry.getDevice(device.getId())
                                            .flatMap(opts -> opts.removeConfigs(productConfiguration.keySet()))
                                            .then(
                                                    //更新数据库
                                                    Mono.just(
                                                            update(
                                                                    (new LambdaUpdateWrapper<IotDevice>())
                                                                            .eq(IotDevice::getId, device.getId())
                                                                            .set(IotDevice::getConfiguration, deviceConfiguration.toString())
                                                            )
                                                    )
                                            );
                                }
                                return Mono.empty();
                            })
                            .thenReturn(deviceConfiguration);
                })
                .defaultIfEmpty(Collections.emptyMap());
    }

    /**
     * 合并更新配置
     *
     * @param deviceId        设备id
     * @param configuration   其他配置信息
     * @param updateOperation 更新条件
     * @return 合并更新结果
     */
    @Override
    public Mono<Void> mergeConfiguration(
            String deviceId,
            Map<String, Object> configuration,
            Function<LambdaUpdateWrapper<IotDevice>, LambdaUpdateWrapper<IotDevice>> updateOperation) {

        if (MapUtils.isEmpty(configuration)) {
            return Mono.empty();
        }

        return Mono.just(getById(deviceId))
                .flatMap(device -> {
                    //合并更新配置
                    device.mergeConfiguration(configuration);

                    LambdaUpdateWrapper<IotDevice> updateWrapper = updateOperation
                            .apply(new LambdaUpdateWrapper<>())
                            .set(IotDevice::getConfiguration, device.getConfiguration())
                            .set(IotDevice::getFeatures, device.getFeatures())
                            .set(IotDevice::getDeriveMetadata, device.getDeriveMetadata())
                            .eq(IotDevice::getId, deviceId);

                    return Mono.just(update(updateWrapper));
                })
                .then(
                        //更新缓存里到信息
                        registry
                                .getDevice(deviceId)
                                .flatMap(device -> device.setConfigs(configuration))
                )
                .then();
    }

    //*********************************设备属性****************************************

    /**
     * 获取设备实时属性值
     *
     * @param deviceId   设备id
     * @param propertyId 属性
     * @return 属性实时数据
     */
    @Override
    public Mono<Map<String, Object>> getProperty(String deviceId, String propertyId, Map<String, Object> headers) {

        return registry
                .getDevice(deviceId)
                .switchIfEmpty(ErrorUtils.notFound("设备未接入"))
                .map(DeviceOperator::messageSender)//发送消息到设备
                .map(sender -> sender
                        .readProperty(propertyId)
                        .headers(headers)
                        .messageId(IDGenerator.SNOW_FLAKE_STRING.generate()))
                .flatMapMany(ReadPropertyMessageSender::send)
                .flatMap(mapReply(ReadPropertyMessageReply::getProperties))
                .reduceWith(LinkedHashMap::new, (main, map) -> {
                    main.putAll(map);
                    return main;
                });
    }

    /**
     * 批量获取设备实时属性
     *
     * @param deviceId   设备id
     * @param properties 属性id集
     * @param headers    消息头
     * @return 属性实时数据
     */
    @Override
    public Mono<Map<String, Object>> getProperties(String deviceId, List<String> properties, Map<String, Object> headers) {
        return registry.getDevice(deviceId)
                .switchIfEmpty(ErrorUtils.notFound("设备未接入"))
                .map(DeviceOperator::messageSender)
                .flatMapMany((sender) ->
                        Mono.just(properties)
                                .flatMapMany(list ->
                                        sender
                                                .readProperty(list.toArray(new String[0]))
                                                .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                                                .headers(headers)
                                                .send()))
                .flatMap(mapReply(ReadPropertyMessageReply::getProperties))
                .reduceWith(LinkedHashMap::new, (main, map) -> {
                    main.putAll(map);
                    return main;
                });
    }


    /**
     * 设备属性写操作
     *
     * @param deviceId 设备id
     * @param props    属性参数
     * @param headers  消息头
     * @return 属性写入操作结果
     */
    @Override
    @SneakyThrows
    public Mono<Map<String, Object>> writeProperties(String deviceId, Map<String, Object> props, Map<String, Object> headers) {
        return registry
                // 获取注册的设备操作对象
                .getDevice(deviceId)
                .switchIfEmpty(ErrorUtils.notFound("设备未接入"))
                .map(operator -> operator
                        // 消息发送器
                        .messageSender()
                        // 修改设备属性消息发送器
                        .writeProperty()
                        // 设置消息id
                        .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                        // 写入属性参数
                        .write(props)
                        .headers(headers)
                )
                // 发送消息
                .flatMapMany(WritePropertyMessageSender::send)
                // 获取消息回复
                .flatMap(mapReply(WritePropertyMessageReply::getProperties))
                // 转换成LinkedHashMap集合返回
                .reduceWith(LinkedHashMap::new, (main, map) -> {
                    main.putAll(map);
                    return main;
                });
    }


    //*********************************设备功能****************************************

    /**
     * 设备功能调用
     *
     * @param deviceId   设备id
     * @param functionId 功能id
     * @param properties 参数
     * @param headers  消息头
     * @return 调用结果
     */
    @Override
    @Deprecated
    public Flux<?> invokeFunction(String deviceId, String functionId, Map<String, Object> properties, Map<String, Object> headers) {
        return registry
                .getDevice(deviceId)
                .switchIfEmpty(ErrorUtils.notFound("设备未接入"))
                .flatMap(operator -> operator
                        .messageSender()
                        .invokeFunction(functionId)
                        .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                        .setParameter(properties)
                        .headers(headers)
                        .validate()
                )
                .flatMapMany(FunctionInvokeMessageSender::send)
                .flatMap(mapReply(FunctionInvokeMessageReply::getOutput));
    }


    //*********************************发送指令*************************************

    /**
     * 发送指令到设备
     *
     * @param deviceId 设备id
     * @param props    发送参数
     * @param headers  消息头
     * @return 指令发送结果
     */
    @Override
    @SuppressWarnings("all")
    public Flux<?> sendMessage(String deviceId, Map<String, Object> props, Map<String, Object> headers) {
        return Mono.just(props)
                .flatMapMany(param -> Mono
                        .zip(
                                // 获取设备发送器
                                registry
                                        // 获取设备操作接口
                                        .getDevice(deviceId)
                                        // 获取消息发送器
                                        .map(DeviceOperator::messageSender)
                                        .switchIfEmpty(Mono.error(() -> new NotFoundException("设备未接入或未激活"))),

                                // 转换消息
                                Mono.<Message>justOrEmpty(MessageType.convertMessage(param))
                                        // 转换消息类型
                                        .cast(DeviceMessage.class)
                                        .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("不支持的消息格式")))

                        ).flatMapMany(tp2 -> {
                            DeviceMessageSender sender = tp2.getT1();
                            DeviceMessage message = tp2.getT2();
                            headers.forEach(message::addHeader);
                            Map<String, String> copy = new HashMap<>();
                            copy.put("deviceId", deviceId);
                            if (!StringUtils.hasText(message.getMessageId())) {
                                copy.put("messageId", IDGenerator.SNOW_FLAKE_STRING.generate());
                            }
                            FastBeanCopier.copy(copy, message);
                            return sender
                                    .send(message)
                                    .onErrorResume(DeviceOperationException.class, error -> {
                                        if (message instanceof RepayableDeviceMessage) {
                                            return Mono.just(
                                                    ((RepayableDeviceMessage) message).newReply().error(error)
                                            );
                                        }
                                        return Mono.error(error);
                                    });
                        }));
    }

    /**
     * 批量发送指令到设备
     *
     * @param deviceIds 设备id
     * @param messages  指令
     * @return 指令发送结果
     */
    @Override
    @SuppressWarnings("all")
    public Flux<?> sendMessage(List<String> deviceIds, List<Map<String, Object>> messages, Map<String, Object> headers) {

        Lazy<Flux<DeviceOperator>> operators = Lazy.of(() -> {

            return Flux.fromIterable(deviceIds)
                    .flatMap(deviceId -> registry.getDevice(deviceId))
                    .cache();
        });

        return Flux.fromIterable(messages)
                .flatMap(message -> {
                    DeviceMessage msg = MessageType
                            // 消息换换
                            .convertMessage(message)
                            .filter(DeviceMessage.class::isInstance)
                            .map(DeviceMessage.class::cast)
                            .orElseThrow(() -> new UnsupportedOperationException("不支持的消息格式:" + message));

                    String deviceId = msg.getDeviceId();
                    Flux<DeviceOperator> devices = StringUtils.isEmpty(deviceId)
                            ? operators.get()
                            : registry.getDevice(deviceId).flux();

                    return devices
                            .flatMap(device -> {
                                Map<String, Object> copy = new HashMap<>(message);
                                copy.put("deviceId", device.getDeviceId());
                                copy.putIfAbsent("messageId", IDGenerator.SNOW_FLAKE_STRING.generate());
                                //复制为新的消息,防止冲突
                                DeviceMessage copiedMessage = MessageType
                                        .convertMessage(copy)
                                        .map(DeviceMessage.class::cast)
                                        .orElseThrow(() -> new UnsupportedOperationException("不支持的消息格式"));
                                headers.forEach(copiedMessage::addHeader);
                                return device
                                        .messageSender()
                                        .send(copiedMessage)
                                        .onErrorResume(Exception.class, error -> {
                                            if (copiedMessage instanceof RepayableDeviceMessage) {
                                                return Mono.just(
                                                        ((RepayableDeviceMessage) copiedMessage).newReply().error(error)
                                                );
                                            }
                                            return Mono.error(error);
                                        });
                            });
                });
    }

    //*********************************物模型*************************************

    /**
     * @param id 设备id
     * @return 获取物模型
     */
    @Override
    public Mono<DeviceMetadata> getMetadata(String id) {
        return registry.getDevice(id)
                .flatMap(DeviceOperator::getMetadata);
    }

    /**
     * 设置当前设备的独立物模型,如果没有设置,这使用产品的物模型配置
     *
     * @param id       设备id
     * @param metadata 物模型
     */
    @Override
    public Mono<Void> updateMetadata(String id, String metadata) {
        return Mono.defer(() -> {
            LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.set(IotDevice::getDeriveMetadata, metadata);
            updateWrapper.eq(IotDevice::getId, id);
            return Mono.just(this.update(updateWrapper));
        })
                .then(registry.getDevice(id))
                .flatMap(device -> device.updateMetadata(metadata))
                .then();
    }

    /**
     * 重置设备物模型，重置后将使用产品物模型配置
     *
     * @param id 设备id
     * @return 重置结果
     */
    @Override
    public Mono<Boolean> resetMetadata(String id) {
        return registry
                .getDevice(id)
                .flatMap(DeviceOperator::resetMetadata)
                .then(Mono.defer(() -> {
                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.set(IotDevice::getDeriveMetadata, null);
                    updateWrapper.eq(IotDevice::getId, id);
                    return Mono.just(this.update(updateWrapper));
                }));
    }

    /**
     * 合并产品的物模型
     *
     * @param id 设备id
     * @return 合并结果
     */
    @Override
    public Mono<Boolean> mergeProductMetadata(String id) {

        return Mono.just(this.getById(id))
                //只有单独保存过物模型的才合并
                .filter(deviceInstance -> StringUtils.hasText(deviceInstance.getDeriveMetadata()))
                .flatMap(deviceInstance -> Mono.just(productService.selectIotProductById(deviceInstance.getProductId()))
                        .flatMap(product -> deviceInstance.mergeMetadata(product.getMetadata()))
                        .then(
                                Mono.defer(() ->
                                        Mono.just(
                                                this.update(
                                                        new LambdaUpdateWrapper<IotDevice>()
                                                                .set(IotDevice::getDeriveMetadata, deviceInstance.getDeriveMetadata())
                                                                .eq(IotDevice::getId, deviceInstance.getId())))
                                                .then(registry.getDevice(deviceInstance.getId()))
                                                .flatMap(device -> device.updateMetadata(deviceInstance.getDeriveMetadata()))
                                )
                        ));

    }


    //*********************************协议支持*************************************

    /**
     * 更新协议支持
     *
     * @param id              设备id
     * @param protocolStandby 协议支持配置
     * @return 更新结果
     */
    @Override
    @Transactional
    public Mono<Boolean> updateProtocolStandby(String id, String protocolStandby) {

        String protocolStandbyValue = IotProduct.toProtocolStandbyValue(protocolStandby);

        return registry.getDevice(id)
                .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("设备[" + id + "]未启用")))
                .flatMap(device -> device.updateProtocolStandby(protocolStandbyValue))
                .then(Mono.defer(() -> {
                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.set(IotDevice::getProtocolStandby, protocolStandby);
                    updateWrapper.eq(IotDevice::getId, id);
                    return Mono.just(update(updateWrapper));
                }));

    }

    /**
     * 重置设备协议支持，重置后将使用产品协议支持
     *
     * @param id 设备id
     * @return 重置结果
     */
    @Override
    public Mono<Boolean> resetProtocolStandby(String id) {

        return registry
                .getDevice(id)
                .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("设备[" + id + "]未启用")))
                .flatMap(DeviceOperator::resetProtocolStandby)
                .then(Mono.defer(() -> {
                    LambdaUpdateWrapper<IotDevice> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.set(IotDevice::getProtocolStandby, null);
                    updateWrapper.eq(IotDevice::getId, id);
                    return Mono.just(update(updateWrapper));
                }));
    }

    /**
     * 生成设备的详情信息
     *
     * @param product 产品信息
     * @param device  设备信息
     * @param tags    标签信息
     * @return 设备详情信息
     */
    protected Mono<DeviceDetail> createDeviceDetail(IotProduct product,
                                                    IotDevice device,
                                                    List<IotDeviceTags> tags) {

        DeviceDetail detail = new DeviceDetail().with(product).with(device).with(tags);
        return Mono
                .zip(
                        //设备信息
                        registry
                                .getDevice(device.getId())
                                .flatMap(operator -> operator
                                        //检查设备的真实状态,可能出现设备已经离线,但是数据库状态未及时更新的.
                                        .checkState()
                                        .map(DeviceState::of)
                                        //检查失败,则返回原始状态
                                        .onErrorReturn(Objects.requireNonNull(DeviceState.of(device.getDeviceState())))
                                        //如果状态不一致,则需要更新数据库中的状态
                                        .filter(state -> state != detail.getState())
                                        .doOnNext(detail::setState)
                                        .flatMap(state -> {
                                            device.setDeviceState(state.getValue());
                                            return Mono.just(this.updateById(device));
                                        })
                                        .thenReturn(operator)),
                        //配置定义
                        metadataManager
                                .getDeviceConfigMetadata(device.getId())
                                .flatMapIterable(ConfigMetadata::getProperties)
                                .collectList(),
                        detail::with
                )
                //填充详情信息
                .flatMap(Function.identity())
                .switchIfEmpty(
                        Mono.defer(() -> {
                            //如果设备注册中心里没有设备信息,并且数据库里的状态不是未激活.
                            //可能是因为注册中心信息丢失,修改数据库中的状态信息.
                            if (detail.getState() != DeviceState.notActive) {
                                device.setDeviceState(DeviceState.notActive.getValue());
                                return Mono.just(this.updateById(device))
                                        .thenReturn(detail.notActive());
                            }
                            return Mono.just(detail.notActive());
                        }).thenReturn(detail))
                .onErrorResume(err -> {
                    log.error("get device detail error", err);
                    return Mono.just(detail);
                });

    }

    private static <R extends DeviceMessageReply, T> Function<R, Mono<T>> mapReply(Function<R, T> function) {
        return reply -> {
            if (ErrorCode.REQUEST_HANDLING.name().equals(reply.getCode())) {
                throw new DeviceOperationException(ErrorCode.REQUEST_HANDLING, reply.getMessage());
            }
            if (!reply.isSuccess()) {
                throw new BusinessException(reply.getMessage(), reply.getCode());
            }
            return Mono.justOrEmpty(function.apply(reply));
        };
    }

    private static <R extends DeviceMessageReply, T> Function<R, T> mapReply2(Function<R, T> function) {
        return reply -> {
            if (ErrorCode.REQUEST_HANDLING.name().equals(reply.getCode())) {
                throw new DeviceOperationException(ErrorCode.REQUEST_HANDLING, reply.getMessage());
            }
            if (!reply.isSuccess()) {
                throw new BusinessException(reply.getMessage(), reply.getCode());
            }
            T mapped = function.apply(reply);
            if (mapped == null) {
                throw new BusinessException(reply.getMessage(), reply.getCode());
            }
            return mapped;
        };
    }

    /**
     * 循环依赖检测器
     */
    private final CyclicDependencyChecker<IotDevice, Void> checker = CyclicDependencyChecker
            .of(IotDevice::getId, IotDevice::getParentId, id -> Mono.just(this.getById(id)));

    /**
     * 检查是否有循环依赖
     *
     * @param id       设备id
     * @param parentId 父设备id
     * @return 检查结果
     */
    public Mono<Void> checkCyclicDependency(String id, String parentId) {
        IotDevice instance = new IotDevice();
        instance.setId(id);
        instance.setProductId(parentId);
        return checker.check(instance);
    }

    /**
     * 删除设备后置处理,解绑子设备和网关,并在注册中心取消激活已激活设备.
     */
    private Flux<Void> deletedHandle(Flux<IotDevice> devices) {
        return devices.filter(device -> !StringUtils.isEmpty(String.valueOf(device.getParentId())))
                .groupBy(IotDevice::getParentId)
                .flatMap(group -> {
                    String parentId = group.key();
                    return group.flatMap(child -> registry.getDevice(child.getId())
                            .flatMap(device -> device.removeConfig(DeviceConfigKey.parentGatewayId.getKey()).thenReturn(device))
                    )
                            .as(childrenDeviceOp -> registry.getDevice(parentId)
                                    .flatMap(gwOperator -> gwOperator.getProtocol()
                                            .flatMap(protocolSupport -> protocolSupport.onChildUnbind(gwOperator, childrenDeviceOp))
                                    )
                            );
                })
                // 取消激活
                .thenMany(
                        devices.filter(device -> !DeviceState.notActive.getValue().equals(device.getDeviceState()))
                                .flatMap(device -> registry.unregisterDevice(device.getId()))
                );
    }

    @EventListener
    public void deletedHandle(EntityDeletedEvent<IotDevice> event) {
        event.async(
                this.deletedHandle(Flux.fromIterable(event.getEntity())).then()
        );
    }


    /**
     * 通过外键值查找本实体，如果查找到则说明本实体关联了外键对应的实体。
     *
     * @param foreignKey      ： 本实体外键属性
     * @param foreignKeyValue ： 本实体的外键值
     * @return
     */
    @Override
    public LinkModel findByForeignKey(String foreignKey, Object foreignKeyValue) {

        if (FOREIGN_KEY_PRODUCT_ID.equals(foreignKey)) {

            LambdaQueryWrapper<IotDevice> queryWrapper = new LambdaQueryWrapper<>();

            queryWrapper.eq(IotDevice::getProductId, foreignKeyValue);

            List<IotDevice> iotCategoryList = list(queryWrapper);

            if (iotCategoryList != null && !iotCategoryList.isEmpty()) {
                return new LinkModel("设备已关联");
            }
        }

        return null;


    }

    /**
     * 根据外键删除所有关联数据
     *
     * @param foreignKey      ： 本实体外键属性
     * @param foreignKeyValue ： 本实体的外键值
     * @return
     */
    @Override
    public boolean deleteAllByForeignKey(String foreignKey, Object foreignKeyValue) {
        return false;
    }

    /**
     * 外键标识
     */
    private static final String FOREIGN_KEY_PRODUCT_ID = "productId";

    /**
     * 如果实体存在外键关联，则向外该键所指的实体服务注册外键，从而建立关联链关系。
     */
    @Override
    public void registerForeignKey() {
        associatedService.registerForeignKey(this, FOREIGN_KEY_PRODUCT_ID);
    }
}
